package com.cn.daimajiangxin.flink;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class SocketWordCount {

    public static void main(String[] args) throws Exception {
        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 启用检查点，确保容错性
        env.enableCheckpointing(10000); // 每10秒创建一次检查点

        // 设置并行度
        env.setParallelism(2);

        // 2. 从Socket读取数据
        String hostname = "192.168.0.199";
        int port = 9999;

        // 支持命令行参数传入
        if (args.length > 0) {
            hostname = args[0];
        }
        if (args.length > 1) {
            port = Integer.parseInt(args[1]);
        }
        DataStream<String> text = env.socketTextStream(
                hostname,
                port,
                "\n", // 行分隔符
                0); // 最大重试次数

        // 3. 数据转换
        DataStream<Tuple2<String, Integer>> wordCounts = text
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((event, timestamp) -> {
                                    String[] parts = event.split(",");
                                    return Long.parseLong(parts[0]);
                                }))
                .flatMap(new Tokenizer())
                .keyBy(value -> value.f0)
                // 添加基于处理时间的滚动窗口计算
                .window(TumblingEventTimeWindows.of(Duration.ofSeconds(5)))
                // 使用sum聚合算子
                .sum(1);

        // 4. 输出结果
        wordCounts.print("Word Count");

        // 5. 启动作业
        env.execute("Socket Word Count");
    }

    // 可选：使用传统的FlatMapFunction实现方式
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        private static final long serialVersionUID = 1L;

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        }
    }
}